package com.chb.flink.func

import java.text.SimpleDateFormat
import java.util.Date

import com.chb.flink.source.StationLog
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.util.Collector

/**
 * 监控每一个手机号，如果在5秒内呼叫它的通话都是失败的，发出警告信息
 * 在5秒中内只要有一个呼叫不是fail则不用警告
 */
object TestProcessFunction {
    def main(args: Array[String]): Unit = {
        //初始化Flink的Streaming（流计算）上下文执行环境
        val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        streamEnv.setParallelism(1)
        //导入隐式转换
        import org.apache.flink.streaming.api.scala._

        //读取socket数据
        val data = streamEnv.socketTextStream("10.0.0.201", 8888)
            .map(line => {
                var arr = line.split(",")
                new StationLog(arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim, arr(4).trim.toLong, arr(5).trim.toLong)
            })

        data.keyBy(_.callOut)
            .process(new MonitorCallFail())
            .print()


        streamEnv.execute()
    }

    class MonitorCallFail() extends KeyedProcessFunction[String, StationLog, String] {
        // 定义一个状态记录时间
        lazy val timeState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("time", classOf[Long]))

        // 处理数据
        override def processElement(value: StationLog,
                                    context: KeyedProcessFunction[String, StationLog, String]#Context,
                                    collector: Collector[String]): Unit = {
            val time = timeState.value() // 从状态中取出时间
            if (value.callType.equals("fail") && time == 0) { // 第一次失败
                // 获取当前时间， 注册定时器
                val now = context.timerService().currentProcessingTime()
                var onTime = now + 5000L // 5秒后触发
                context.timerService().registerProcessingTimeTimer(onTime);
                println("first time: " + new Date())
                timeState.update(onTime)
            }

            // 有呼叫成功， 取消触发器
            if (!value.callType.equals("fail") && time != 0) {
                context.timerService().deleteProcessingTimeTimer(time)
                timeState.clear()
            }


        }

        // 时间到， 执行触发器，发出告警
        override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, StationLog, String]#OnTimerContext,
                             out: Collector[String]): Unit = {
            val df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
            var warnStr = "触发时间：" + df.format(new Date(timestamp)) + " 手机号：" + ctx.getCurrentKey
            out.collect(warnStr)
            timeState.clear()
        }

    }



}
